# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: helpers\issue_tracker\clients\jira_client.py
import json, html
from . import util
from ..exceptions import ConfigError
from .base_client import BaseClient
from requests.auth import HTTPBasicAuth


class JiraClient(BaseClient):
    """
    JiraClient handles all communication between Acunetix and a specified JIRA instance.
    
    The current client implementation works on version 2 of JIRA's API and
     supports the following functionality:
        - Authentication
            (a). Cookie-Based
            (b). Basic Authentication
    
            .. note: No current support for OAuth1/2 authentication will be implemented
        - Get Projects(*)
        - Get Issue Status(*)
        - Get Issue Types
            + For all projects
            + For specified project
        - Create Issue
            .. note: Creating bulk issues currently not supported, may be implemented
    
        * In the context of the authenticated user
    
    Instantiation of the JIRA client should follow the following structure:
    
    >>> import helpers.issue_tracker.clients.client as client
    >>>
    >>> target_config = {
    >>>     "bug_tracker": "jira",
    >>>     "url": "http://localhost:9000/",
    >>>     "auth": {
    >>>         "kind": "cookie",
    >>>         "user": "root",
    >>>         "password": "toor"
    >>>     },
    >>>     "project": 'Testing',
    >>>     "issuetype": '10100'  # Mandatory
    >>> }
    >>> client = client.get_client(target_config)
    
    See the :func:`establish_connection` as to how establishing a connection works
    See the :func:`get_client` in :file:`client.py` as to how the factory function works
    """

    def __init__(self, target_config, proxies_cb=None, ssl_ca_bundle_path=True):
        super().__init__(target_config, proxies_cb, ssl_ca_bundle_path)
        self.metadata = self._JiraClient__get_metadata()

    def __get_metadata(self):
        """
        Pull metadata for that JIRA instance, mainly only used by JIRA.
        
        Metadata is used and even required by JIRA in order to be able to create and push
         issues to JIRA; including information such as issue types for each project including
         their respective IDs.
        
        This method is only used internally by this class.
        """
        return (
            super()._request(
                endpoint="/rest/api/2/issue/createmeta?expand=projects.issuetypes.fields",
                data=None,
            )
        ).json()

    def _format_vulnerability_description(self, description):
        """Format Acunetix HTML to JIRA's markdown flavour
        
        .. note::
        Currently works by using string swaps rather than regex. Tested this
        using regex however was visibly slower when working with large
        descriptions and the code itself was a big eyesore.
        
        
        :param description: Acunetix Vulnerability description in HTML format
        :return: Acunetix Vulnerability description in JIRA Markdown format
        """
        tags = [
            ("<i>", "_"),
            ("</i>", "_"),
            ("<code><pre>", "{code}"),
            ("</pre></code>", "{code}"),
            ("<pre><code>", "{code}"),
            ("</code></pre>", "{code}"),
            ("<br/>", "\\\\ "),
            ("<hr>", "\n----\n"),
            ("<strong>", "*"),
            ("</strong>", "*"),
            ("<p>", " "),
            ("</p>", ""),
            ("<h1>", "\nh1. "),
            ("</h1>", "\n"),
            ("<h2>", "\nh2. "),
            ("</h2>", "\n"),
            ("<h3>", "\nh3. "),
            ("</h3>", "\n"),
            ("<h4>", "\nh4. "),
            ("</h4>", "\n"),
            ("<h5>", "\nh5. "),
            ("</h5>", "\n"),
            ("<h6>", "\nh6. "),
            ("</h6>", "\n"),
            ("<code>", "{code}"),
            ("</code>", "{code}"),
            ('<span class="bb-dark">', "{color:black}"),
            ("</span>", "{color}"),
            ('<div class="bb-coolbox">', "\nbq. "),
            ("</div>", ""),
            ("<ol>", ""),
            ("</ol>", ""),
            ("<ul>", ""),
            ("</ul>", ""),
            ("<li>", "\n- "),
            ("</li>", ""),
            ("<pre>", "{noformat}"),
            ("</pre>", "{noformat}"),
        ]
        for tag in tags:
            description = description.replace(tag[0], tag[1])

        return description

    def _format_vulnerability_references(self, references):
        """
        Format Acunetix vulnerability references into JIRA links.
        
        All references must be grouped in a single-element list or
         double-element list in order to function.
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "jira",
        >>>     "url": "http://localhost:9000/",
        >>>     "auth": {
        >>>         "kind": "cookie",
        >>>         "user": "root",
        >>>         "password": "toor"
        >>>     },
        >>>     "project": 'Testing',
        >>>     "issuetype": '10100'  # Mandatory
        >>> }
        >>>
        >>> references = [['Acunetix SQL Injection Attack', 'http://www.acunetix.com/...'],
        >>>               ['VIDEO: SQL Injection tutorial', 'http://www.acunetix.com/blog/...']]
        >>>
        >>> jira_client = client.get_client(target_config)
        >>> jira_client._format_vulnerability_references(references)
        >>> ['[Acunetix SQL Injection Attack|http://www.acunetix.com/...]',
        >>>  '[VIDEO: SQL Injection tutorial|http://www.acunetix.com/blog/...]']
        """
        html_references = []
        try:
            for pair in references:
                if not isinstance(pair, (list, tuple)):
                    raise AssertionError("Invalid type, must be of type list or tuple")
                if len(pair) == 1:
                    html_references.append("[" + pair[0] + "]")
                elif len(pair) == 2:
                    html_references.append("[" + pair[0] + "|" + pair[1] + "]")
                else:
                    raise ValueError(
                        "Reference pair has too many or too few"
                        + "items to unpack, must be in pairs of "
                        + "1 or 2"
                    )

        except (AttributeError, TypeError):
            raise AssertionError("References must be an iterable object")

        return html_references

    def _establish_session(self):
        """
        Establish new user session with specified JIRA instance.
        
        .. note::
        
        JIRA API currently supports:
         - Cookie-based Authentication
         - HTTP Basic Authentication
         - OAuth2 Authentication
        
        Current JIRAClient supports Cookie-based and HTTP Basic authentication.
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "jira",
        >>>     "url": "http://localhost:9000/",
        >>>     "auth": {
        >>>         "kind": "cookie",
        >>>         "user": "root",
        >>>         "password": "toor"
        >>>     },
        >>>     "project": 'acx',
        >>>     "issuetype": '10100'  # Mandatory
        >>> }
        >>>
        >>> jira_client = client.get_client(target_config)
        
        
        :return: True if authentication successful, false if otherwise.
        """
        username = self.target_config["auth"]["user"]
        password = self.target_config["auth"]["password"]
        auth_method = self.target_config["auth"]["kind"]
        if auth_method == "cookie":
            credentials = json.dumps(
                {"username": username, "password": password}
            ).encode("utf-8")
            request = super()._request(
                endpoint="/rest/auth/1/session", data=credentials
            )
        else:
            if auth_method == "http_basic":
                request = super()._request(
                    endpoint="/rest/auth/1/session",
                    auth=HTTPBasicAuth(username, password),
                )
            else:
                raise ConfigError(
                    "Unknown Authentication method parsed - Must be 'cookie' or 'http_basic'."
                )
            if request.status_code != 200:
                raise ConfigError(
                    "Invalid login credentials, account locked out or CAPTCHA required."
                )
        try:
            if "application/json" not in request.headers["Content-Type"]:
                if "loginInfo" not in request.json():
                    if "atlassian.xsrf.token" not in request.headers["Set-Cookie"]:
                        raise KeyError
        except KeyError:
            raise ConfigError(
                "Unexpected response returned. Normally caused by an invalid URL."
            )

    def add_vulnerability(self, vuln_data):
        """POST new issue into JIRA instance with specified data.
        
        Vulnerability description between HTML to JIRA format is
        handled automatically. Do NOT modify the Acunetix vulnerability
        descriptions.
        
        :Example:
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "jira",
        >>>     "url": "http://localhost:9000/",
        >>>     "auth": {
        >>>         "kind": "cookie",
        >>>         "user": "root",
        >>>         "password": "toor"
        >>>     },
        >>>     "project": 'Testing',
        >>>     "issuetype": '10100'  # Mandatory
        >>> }
        >>>
        >>> # Sample data provided by Acunetix
        >>> vuln_data = {
        >>>    'vt_name': 'foo',
        >>>    'long_description': '',
        >>>    'description': 'Test Description',
        >>>    'details': 'Attack Details',
        >>>    'vt_tags': ['sql_injection', 'CWE-89']
        >>> }
        >>>
        >>> jira_client = client.get_client(target_config)
        >>> jira_client.add_vulnerability(vuln_data)
        
        :param vuln_data: Issue data POSTed into JIRA. Format:
                            "fields": {
                                "project": {},
                                "summary": "",
                                "issuetype": {},
                                "description": "",
                                "assignee": {},
                                "reporter": {},
                                "priority": {},
                                "labels": []
                            }
        :return: New Issue ID if successfully created (201), otherwise None.
        """
        vuln_description = vuln_data["description"]
        attack_details = vuln_data["details"]
        request_data = vuln_data.get("request")
        if request_data:
            request_details = "<pre>" + request_data + "</pre>"
        else:
            request_details = ""
        target_information = (
            "| <strong>Target URL</strong> | "
            + vuln_data.get("target_address")
            + "|\n"
            + "| <strong>Severity</strong> | "
            + util.evaluate_severity(vuln_data.get("severity"))
            + "|\n"
        )
        recommendation = (
            vuln_data["recommendation"] if vuln_data["recommendation"] else ""
        )
        impact = vuln_data["impact"] if vuln_data["impact"] else ""
        description = self._format_vulnerability_description(
            target_information
            + "<h5>Vulnerability Description</h5>"
            + vuln_description
            + "\n<h5>Attack Details</h5>"
            + attack_details
            + "\n<h5>HTTP Request</h5>"
            + request_details
            + "\n<h5>Impact</h5>"
            + impact
            + "\n<h5>Remediation</h5>"
            + recommendation
        )
        reference_tags = self._format_vulnerability_references(vuln_data["refs"])
        if len(reference_tags) > 0:
            description += "\n<hr><h5>References</h5>"
            for reference_tag in reference_tags:
                description += reference_tag + "<br/>"

            description += "<br/>"
        description = html.unescape(description)
        project_id = None
        if self.get_projects():
            for project in self.get_projects():
                if (
                    project["project_name"].lower()
                    == self.target_config["project"]["project_name"].lower()
                ):
                    project_id = project["project_id"]

            if not project_id:
                raise ValueError(
                    "Unable to resolve project ID. This normally occurs when the user-specified "
                    + "project name is either invalid or does not exist for that logged in user."
                )
        post_data = {
            "fields": {
                "project": {"id": project_id},
                "summary": "Acunetix - " + vuln_data["vt_name"],
                "issuetype": {"id": self.target_config["issue_type"]["issue_id"]},
                "description": self._format_vulnerability_description(description),
            }
        }
        labels_allowed = False
        for project in self.metadata["projects"]:
            if project["name"] == self.target_config["project"]["project_name"]:
                for issuetype in project["issuetypes"]:
                    if issuetype["id"] == self.target_config["issue_type"]:
                        labels_allowed = "labels" in issuetype["fields"]

        if labels_allowed:
            labels = super()._format_vulnerability_tags(vuln_data["vt_tags"])
            post_data["fields"]["labels"] = labels
        request = super()._request(
            endpoint="/rest/api/2/issue", data=json.dumps(post_data).encode("utf-8")
        )
        try:
            if request.status_code == 201:
                return int(request.json()["id"])
            raise Exception(
                "Invalid status code returned. Unable to add issue successfully."
            )
        except KeyError:
            raise Exception(
                "Unexpected response body returned, unable to parse issue number."
            )

    def get_projects(self):
        """Get list of projects available to authenticated user.
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "jira",
        >>>     "url": "http://localhost:9000/",
        >>>     "auth": {
        >>>         "kind": "cookie",
        >>>         "user": "root",
        >>>         "password": "toor"
        >>>     },
        >>>     "project": 'Testing',
        >>>     "issuetype": '10100'  # Mandatory
        >>> }
        >>>
        >>> jira_client = client.get_client(target_config)
        >>> jira_client.get_projects()
        >>> ['Test', 'ProjectA']
        
        :return: List of projects available for the authenticated user
        """
        request = super()._request(endpoint="/rest/api/2/project/")
        if request.status_code == 200:
            try:
                projects = request.json()
                return [
                    {"project_id": str(project["id"]), "project_name": project["name"]}
                    for project in projects
                ]
            except KeyError:
                raise Exception(
                    "Received unexpected response body, unable to parse project name and ID."
                )

            raise Exception(
                "Request to JIRA unsuccessful. Unable to retrieve list of projects."
            )

    def get_issue_types(self, repository=None):
        """Extract issue-types for each project or specified project.
        
        .. note:: In the case of JIRA this will ONLY return the standard
                    issue-types and will omit any subtask issue types.
        
        Format is returned in { "projects": [
                                    project: {
                                        issuetypes: []
                                    }
                                 ]
                              }
        
        :return: List of issues types per project for authenticated user.
        """
        repository = self.target_config["project"]["project_name"]
        projects = {repository: None}
        for project in self.metadata["projects"]:
            if project["name"] == repository:
                projects = [
                    {"issue_id": issuetype["id"], "issue_name": issuetype["name"]}
                    for issuetype in project["issuetypes"]
                    if not issuetype["subtask"]
                ]

        return projects

    def get_issue_status(self, issue_id):
        """Get the status of a specific issue (e.g. In Progress, Done)
        
        >>> import helpers.issue_tracker.clients.client as client
        >>>
        >>> target_config = {
        >>>     "bug_tracker": "jira",
        >>>     "url": "http://localhost:9000/",
        >>>     "auth": {
        >>>         "kind": "cookie",
        >>>         "user": "root",
        >>>         "password": "toor"
        >>>     },
        >>>     "project": 'Testing',
        >>>     "issuetype": '10100'  # Mandatory
        >>> }
        >>>
        >>> jira_client = client.get_client(target_config)
        >>>
        >>> jira_client.get_issue_status(1)
        >>> 'Done'
        >>> jira_client.get_issue_status(-1)
        >>> None
        
        :param issue_id: ID of issue to extract status from
        :return: Status of the specified issue
        """
        request = super()._request(endpoint="/rest/api/2/issue/" + str(issue_id))
        try:
            if request.status_code == 200:
                return request.json()["fields"]["status"]["name"]
        except KeyError:
            raise Exception("Unexpected response body returned")

        raise Exception("Unable to retrieve issue status for issue: " + str(issue_id))
